package com.atguigu.realtime.app.dwd;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONAware;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.realtime.app.BaseAppV1;
import com.atguigu.realtime.common.Constant;
import com.atguigu.realtime.util.FlinkSinkUtil;
import com.atguigu.realtime.util.KKutil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;
import java.util.*;

public class DwdLogApp extends BaseAppV1 {
    private static final String PAGE = "page";
    private static final String DISPLAY = "display";
    private static final String START = "start";

       public static void main(String[] args) {

           new DwdLogApp().init(2001,1,"DwdLogApp","DwdLogApp", Constant.TOPIC_ODS_LOG);
           }


    @Override
    protected void run(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
           //对流进行处理

        //1.纠正新老客户的标记
        SingleOutputStreamOperator<JSONObject> validateStream = validateNewOrOld(stream);

        // 2.分流： 页面  曝光 启动
        HashMap<String, DataStream<JSONObject>> threeStream = splitStream(validateStream);


        // 3.不同的流写入不同的topic中
        write2Kafka(threeStream);

    }

  //  2 two to 4 four for  i18n 国际化
    private void write2Kafka(HashMap<String, DataStream<JSONObject>> threeStream) {

           threeStream
                   .get(PAGE)
                   //转成字符串格式
                   .map(JSONAware::toJSONString)
                   .addSink(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_PAGE));
        threeStream
                .get(START)
                //转成字符串格式
                .map(obj -> obj.toJSONString())
                .addSink(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_START));
        threeStream
                .get(DISPLAY)
                //转成字符串格式
                .map(obj -> obj.toJSONString())
                .addSink(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_DISPLAY));


    }

    private HashMap<String, DataStream<JSONObject>> splitStream(SingleOutputStreamOperator<JSONObject> stream) {
        // 2.分流： 页面  曝光 启动
         OutputTag<JSONObject> pageTag = new OutputTag<JSONObject>(PAGE) {};
         OutputTag<JSONObject> displayTag = new OutputTag<JSONObject>(DISPLAY) {};

        SingleOutputStreamOperator<JSONObject> startStream = stream
                .process(new ProcessFunction<JSONObject, JSONObject>() {

                    @Override
                    public void processElement(JSONObject value,
                                               Context ctx,
                                               Collector<JSONObject> out) throws Exception {
                        // 启动：主流   曝光：侧输出流  页面： 侧输出流
                        JSONObject start = value.getJSONObject("start");
                        if (start != null) {
                            //启动日志
                            out.collect(value);
                        } else  {
                            // 同一条日志极可能即使曝光又是页面
                            JSONObject page = value.getJSONObject("page");
                            if (page != null) {
                                //侧输出流
                                ctx.output(pageTag,value);
                            }
                            JSONArray displays = value.getJSONArray("displays");
                            if (displays != null) {
                                // 把曝光日志拍平 ，数组又多少记录，就向流中放入多少记录
                                Long ts = value.getLong("ts");
                                JSONObject common = value.getJSONObject("common");
                                JSONObject page1 = value.getJSONObject("page");

                                for (int i = 0; i < displays.size(); i++) {
                                    JSONObject display = displays.getJSONObject(i);
                                    // display中添加一些元素，因为缺少一些必要的元素
                                     display.put("ts", ts);
                                     // 仍入集合
                                     display.putAll(common);
                                     display.putAll(page1);


                                    ctx.output(displayTag,display);

                                }
                            }
                        }
                    }
                });
        DataStream<JSONObject> pageStream = startStream.getSideOutput(pageTag);
        DataStream<JSONObject> displaytream = startStream.getSideOutput(displayTag);
        // 返回三个流
        // 返回Tuple3
        // 返回list
        // 返回map
        HashMap<String, DataStream<JSONObject>> map = new HashMap<>();
        map.put(START,startStream);
        map.put(DISPLAY,displaytream);
        map.put(PAGE,pageStream);
        return map;

    }

    private SingleOutputStreamOperator<JSONObject> validateNewOrOld(DataStreamSource<String> stream) {
        /*
        * 如判断一个用户是否为 新老用户？
        *
        * 使用状态
        *
        * 时间事件  加  窗口
        *
        * 第一个窗口 中 时间戳最小的是 is_new 是 1  其他的都应该是0
         *
        * */
       return stream
              //  .map(log -> {return JSON.parseObject(log)}) 简化
                .map(JSON::parseObject)
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        //提起事件时间
                        .withTimestampAssigner((obj,ts) -> obj.getLong("ts"))
                )
                .keyBy(obj -> obj.getJSONObject("common").getString("mid"))
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .process(new ProcessWindowFunction<JSONObject, JSONObject, String, TimeWindow>() {
                    private ValueState<Boolean> firstWindowState;
                    // 因为只需要修改 is_new ,所以返回相同类型

                    //通过open实现查找键控状态
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        firstWindowState = getRuntimeContext().getState(new ValueStateDescriptor<Boolean>("firstWindowState", Boolean.class));
                    }

                    @Override
                    public void process(String key,
                                        Context ctx,
                                        Iterable<JSONObject> elements,
                                        Collector<JSONObject> out) throws Exception {
                        if (firstWindowState.value() == null) {
                            firstWindowState.update(true);
                            List<JSONObject> list = KKutil.toList(elements);
                         //   Collections.min(list,((o1, o2) -> o1.getLong("ts").compareTo(o2.getLong("ts"))) )
                            JSONObject min = Collections.min(list, (Comparator.comparing(o -> o.getLong("ts"))));
                            for (JSONObject obj : list) {
                                if (obj == min) { //判断对象是否相等
                                    obj.getJSONObject("common").put("is_new","1");
                                }else {
                                    obj.getJSONObject("common").put("is_new","0");
                                }
                                out.collect(obj);
                            }

                        }else {
                            //不是第一个窗口  所有元素的is_new  全部为0
                            for (JSONObject obj : elements) {
                                obj.getJSONObject("common").put("is_new","0");
                                out.collect(obj);
                            }
                        }

                   }
                });


    }
}
